/**
 * CPAR TOOLKIT - Compiler and Runtime for CPAR
 * Copyright (C) 2012 CPAR TOOLKIT Project Developers
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
#include <stdlib.h>
#include <memory.h>
#include "cpar.h"

static cpar_status CPARConcurrentQueueExpand(CPARConcurrentQueue*);

struct CPARConcurrentQueue
{
  CPAREvent       *waitEvent;

  void            **items;
  long            itemCount;
  long            maxItems;

  long            readIndex;
  long            writeIndex;

  CPARMutex       *lock;

  CPARAllocator   *allocator;
  long            allocatorIdx;
};

static
cpar_status
CPARConcurrentQueueExpand(CPARConcurrentQueue *queue)
{
  void  **items;
  long  itemCount;

  itemCount = queue->maxItems + 10;
  items = (void**)CPARAllocatorAlloc(queue->allocator, sizeof(void*) * itemCount);
  if(NULL == items)
    return CPAR_E_NO_MEMORY;

  if(queue->items)
  {
    memcpy(items,queue->items,sizeof(void*) * queue->itemCount);
    CPARAllocatorFree(queue->allocator, queue->items);
  }

  queue->items    = items;
  queue->maxItems += itemCount;

  return CPAR_SUCCESS;
}

static
void
CPARConcurrentQueueAllocatorListener(CPARAllocator *allocator, void *data)
{
  CPARConcurrentQueue *queue;

  queue = (CPARConcurrentQueue*)data;
  CPARConcurrentQueueDestroy(queue);
}

cpar_status CPARConcurrentQueueCreate( CPARAllocator *allocator, CPARConcurrentQueue **queue )
{
  cpar_status s;
  CPARConcurrentQueue *p;

  p = (CPARConcurrentQueue*) CPARAllocatorAlloc(allocator, sizeof(CPARConcurrentQueue));
  if(NULL == p)
    return CPAR_E_NO_MEMORY;

  p->waitEvent  = NULL;
  p->lock       = NULL;

  p->readIndex  = 0;
  p->writeIndex = 0;
  p->itemCount  = 0;
  p->items      = NULL;
  p->maxItems   = 0;

  p->allocator    = allocator;
  p->allocatorIdx = CPARAllocatorDestroyListenerAdd(allocator, CPARConcurrentQueueAllocatorListener, p);
  
  s = CPAREventCreate(allocator, &(p->waitEvent));
  if(s)
  {
    CPARConcurrentQueueDestroy(p);
    return s;
  }

  s = CPARMutexCreate(allocator, &(p->lock));
  if(s)
  {
    CPARConcurrentQueueDestroy(p);
    return s;
  }

  *queue = p;
  
  return CPAR_SUCCESS;  
}

void CPARConcurrentQueueDestroy( CPARConcurrentQueue *queue )
{
  if(queue->waitEvent)
    CPAREventDestroy(queue->waitEvent);

  if(queue->lock)
    CPARMutexDestroy(queue->lock);

  CPARAllocatorFree(queue->allocator,queue);
}

long CPARConcurrentQueueCount( CPARConcurrentQueue *queue )
{
  return queue->itemCount;
}

cpar_status
CPARConcurrentQueueEnqueue( CPARConcurrentQueue *queue, void *item )
{
  cpar_status s;
  long        cnt;

  CPARMutexLock(queue->lock);

  if(queue->itemCount == queue->maxItems)
  {
    s = CPARConcurrentQueueExpand( queue );
    if(s)
      return s;
  }

  queue->items[queue->writeIndex] = item;  
  
  queue->writeIndex++;

  cnt = queue->itemCount++;

  CPARMutexUnlock(queue->lock);
  
  if(cnt == 1)
    CPAREventSet(queue->waitEvent);

  return CPAR_SUCCESS;
}

static
void*
_CPARConcurrentQueueDequeue(CPARConcurrentQueue *queue)
{
  void  *pData;

  pData = NULL;
  CPARMutexLock(queue->lock);
  if(queue->itemCount > 0)
  {
    queue->itemCount--;
    pData = queue->items[queue->readIndex];
    queue->readIndex++;
    if(queue->readIndex == queue->writeIndex)
      queue->readIndex = queue->writeIndex = 0;
  }
  CPARMutexUnlock(queue->lock);
  return pData;
}

void *CPARConcurrentQueueDequeue( CPARConcurrentQueue *queue )
{
  void  *pData;

  pData = NULL;
  while(NULL == pData)
  {
    pData = _CPARConcurrentQueueDequeue(queue);
    if(NULL == pData)
    {
      CPAREventWait(queue->waitEvent);
      continue;
    }
  }
  return pData;
}

void*
CPARConcurrentQueueTryDequeue( CPARConcurrentQueue *queue, long milliseconds )
{
  void  *pData;

  pData = _CPARConcurrentQueueDequeue(queue);
  if(NULL == pData)
  {
    if(!CPAREventTryWait(queue->waitEvent, milliseconds))
      return NULL;

    return _CPARConcurrentQueueDequeue(queue);
  }
  return pData;
}